package com.coolw.test.delaymessage;

import lombok.Data;
import lombok.NoArgsConstructor;
import org.dromara.hutool.core.data.id.IdUtil;

import java.io.Serializable;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * DelayQueue 延迟消息
 *
 * @author coolw
 * @date 2025/2/28 10:57
 */
public class DelayQueueDemo {

    public static void main(String[] args) {
        DelayQueue<DelayedTask<Order>> queue = new DelayQueue<>();

        // 开启一个线程来生产消息，延迟5S
        new Thread(new Producer(queue, 5)).start();

        // 开启一个线程来消费消息
        new Thread(new Consumer(queue)).start();
    }

    /**
     * 消息生产者
     */
    static class Producer implements Runnable {

        private final DelayQueue<DelayedTask<Order>> queue;

        private final int delaySecond;

        public Producer(DelayQueue<DelayedTask<Order>> queue, int delaySecond) {
            this.queue = queue;
            this.delaySecond = delaySecond;
        }

        @Override
        public void run() {
            int count = 1;
            String orderId = "test-001-";
            while (true) {
                for (int i = 1; i <= 5; i++) {
                    DelayedTask<Order> task = new DelayedTask<>(this.delaySecond, "order", new Order(orderId + count));
                    queue.add(task);
                    System.out.println("=====》生产延迟消息,task:" + task);
                    count++;
                }

                // 阻塞,测试看效果
                try {
                    Thread.sleep(5000L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 消息消费者
     */
    static class Consumer implements Runnable {

        private final DelayQueue<DelayedTask<Order>> queue;

        public Consumer(DelayQueue<DelayedTask<Order>> queue) {
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true) {
                DelayedTask<Order> task = null;
                try {
                    task = queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (task != null) {
                    System.out.println("《=====消费延迟消息,task:" + task);
                } else {
                    System.out.println("消费延迟消息失败,消息内容为空");
                }
            }
        }
    }

    static class DelayedTask<T> implements Delayed {

        /**
         * 任务超时时间戳
         */
        private final long expire;
        private final String messageId;
        private final String messageType;
        private final T data;

        public DelayedTask(int delaySecond, String messageType, T data) {
            // 任务超时时间戳 = 当前时间 + 延迟时间
            this.expire = CalendarUtil.getTimeMillis(delaySecond);
            this.messageId = IdUtil.fastSimpleUUID();
            this.messageType = messageType;
            this.data = data;
        }

        /**
         * 获取此对象剩余的延迟时间（过期时间-当前时间）
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return expire - CalendarUtil.getInstance().getTimeInMillis();
        }

        /**
         * 比较排序（当前时间的延迟时间 - 比对对象的延迟时间）
         */
        @Override
        public int compareTo(Delayed o) {
            long d = getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
            return d == 0 ? 0 : (d < 0 ? -1 : 1);
        }

        @Override
        public String toString() {
            return "DelayedTask{" +
                    "messageId='" + messageId + '\'' +
                    ", messageType=" + messageType +
                    ", data=" + data +
                    '}';
        }
    }

    @NoArgsConstructor
    @Data
    static class Order implements Serializable {
        private static final long serialVersionUID = 2903688150916823637L;

        private String orderId;

        public Order(String orderId) {
            this.orderId = orderId;
        }
    }
}
